package com.ruoyi.web.controller.kafka;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.ruoyi.common.utils.ToolUtil;
import com.ruoyi.web.controller.test.TestKafka;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @Description:
 * @Author: LJ
 * @Date: 2022/9/23 11:15
 * @version:
 */

@Component
@Slf4j
public class KafkaConsumer {
    @KafkaListener(topicPattern = KafkaProducer.TOPIC_TEST, concurrency = "5", groupId = "scrm")
    public void topic_test(ConsumerRecord<String, String> record, Acknowledgment ack) {
        try {
            TestKafka testKafka = jsonMessageToObject(record, TestKafka.class);
            log.info("topic_test 消费了： Topic:" + record.topic() + ",Message:" + JSONObject.toJSONString(testKafka));
        } catch (Exception e) {
            log.error("topic_test 消息接收失败了： Topic:" + record.topic() + ",Message:" + record.value());
        } finally {
            ack.acknowledge();
        }
    }


    /**
     * json 转 <T>
     */
    private <T> T jsonMessageToObject(ConsumerRecord<String, String> cr, Class<T> clazz) {
        log.debug("收到消息Kafka receive <<- {}", cr.topic());
        String value = cr.value();
        if (ToolUtil.isEmpty(value)) return null;
        return JSON.parseObject(value, clazz);
    }

    private Map jsonMessageToMap(ConsumerRecord<String, String> cr) {
        log.debug("收到消息Kafka receive <<- {}", cr.topic());
        String value = cr.value();
        if (ToolUtil.isEmpty(value)) return new HashMap();
        return (Map) JSON.parse(value);
    }

    /**
     * json 转 List<T>
     */
    private <T> List<T> jsonMessageToList(ConsumerRecord<String, String> cr, Class<T> clazz) {
        log.debug("收到消息Kafka receive <<- {}", cr.topic());
        String value = cr.value();
        if (ToolUtil.isEmpty(value)) return null;
        return JSONArray.parseArray(value,clazz);
    }

    /**
     * json 转 Map
     */
    private Map jsonMessageToList(ConsumerRecord<String, String> cr) {
        log.debug("收到消息Kafka receive <<- {}", cr.topic());
        String value = cr.value();
        if (ToolUtil.isEmpty(value)) return new HashMap();
        return (Map) JSON.parse(value);
    }
}
